#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright 2019 Mobvoi Inc. All Rights Reserved.
# Author: di.wu@mobvoi.com (DI WU)
"""Subsampling layer definition."""

from typing import Tuple

import torch

from wenet.models.moe_comformer.download_simple import SpeechProjectionModule


class BaseSubsampling(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.right_context = 0
        self.subsampling_rate = 1

    def position_encoding(self, offset: int, size: int) -> torch.Tensor:
        return self.pos_enc.position_encoding(offset, size)


class LinearNoSubsampling(BaseSubsampling):
    """Linear transform the input without subsampling

    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.

    """
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, **kwargs):
        """Construct an linear object."""
        super().__init__()
        self.out = torch.nn.Sequential(
            torch.nn.Linear(idim, odim),
            torch.nn.LayerNorm(odim, eps=1e-5),
            torch.nn.Dropout(dropout_rate),
        )
        self.pos_enc = pos_enc_class
        self.right_context = 0
        self.subsampling_rate = 1

    def forward(
            self,
            x: torch.Tensor,
            x_mask: torch.Tensor,
            offset: int = 0
    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Input x.

        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: linear input tensor (#batch, time', odim),
                where time' = time .
            torch.Tensor: linear input mask (#batch, 1, time'),
                where time' = time .

        """
        x = self.out(x)
        x, pos_emb = self.pos_enc(x, offset)
        return x, pos_emb, x_mask


class Conv2dSubsampling4(BaseSubsampling):
    """Convolutional 2D subsampling (to 1/4 length).

    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.

    """
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, norm: bool = False, **kwargs):
        """Construct an Conv2dSubsampling4 object."""
        super().__init__()
        self.conv = torch.nn.Sequential(
            torch.nn.Conv2d(1, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
        )
        self.linear = torch.nn.Linear(odim * ((idim - 3) // 4), odim)
        self.pos_enc = pos_enc_class
        self.norm = torch.nn.LayerNorm(odim) if norm else torch.nn.Identity()
        # The right context for every conv layer is computed by:
        # (kernel_size - 1) * frame_rate_of_this_layer
        self.subsampling_rate = 4
        # 6 = (3 - 1) * 1 + (3 - 1) * 2
        self.right_context = 6

    def forward(self, x: torch.Tensor, x_mask: torch.Tensor, offset: int = 0
               ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.

        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // 4.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // 4.
            torch.Tensor: positional encoding

        """
        x = x.unsqueeze(1)  # (b, c=1, t, f)
        x = self.conv(x)
        b, c, t, f = x.size()
        x = self.linear(x.transpose(1, 2).contiguous().view(b, t, c * f))
        x, pos_emb = self.pos_enc(x, offset)
        return self.norm(x), pos_emb, x_mask[:, :, 6::4]


class Conv2dSubsampling6(BaseSubsampling):
    """Convolutional 2D subsampling (to 1/6 length).
    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.
        pos_enc (torch.nn.Module): Custom position encoding layer. ---; ---; ---; ---; ---;
    """
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, norm: bool = False, **kwargs):
        """Construct an Conv2dSubsampling6 object."""
        super().__init__()
        self.conv = torch.nn.Sequential(
            torch.nn.Conv2d(1, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 5, 3),
            torch.nn.ReLU(),
        )
        self.linear = torch.nn.Linear(odim * ((idim - 5) // 6), odim)
        self.pos_enc = pos_enc_class
        self.norm = torch.nn.LayerNorm(odim) if norm else torch.nn.Identity()
        # 10 = (3 - 1) * 1 + (5 - 1) * 2
        self.subsampling_rate = 6
        self.right_context = 10

    def forward(self, x: torch.Tensor, x_mask: torch.Tensor, offset: int = 0
               ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.
        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // 6.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // 6.
            torch.Tensor: positional encoding
        """
        x = x.unsqueeze(1)  # (b, c, t, f)
        x = self.conv(x)
        b, c, t, f = x.size()
        x = self.linear(x.transpose(1, 2).contiguous().view(b, t, c * f))
        x, pos_emb = self.pos_enc(x, offset)
        return self.norm(x), pos_emb, x_mask[:, :, 10::6]

class Conv2dSubsampling6_simple2(BaseSubsampling):
    """Convolutional 2D subsampling (to 1/6 length).
    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.
        pos_enc (torch.nn.Module): Custom position encoding layer. ---; ---; ---; ---; ---;
    """
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, norm: bool = False, **kwargs):
        """Construct an Conv2dSubsampling6 object."""
        super().__init__()
        self.conv = torch.nn.Sequential(
            torch.nn.Conv2d(1, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 5, 3),
            torch.nn.ReLU(),
        )
        self.linear = torch.nn.Linear(odim * ((idim - 5) // 6), odim)
        self.pos_enc = pos_enc_class
        self.norm = torch.nn.LayerNorm(odim) if norm else torch.nn.Identity()
        # 10 = (3 - 1) * 1 + (5 - 1) * 2
        self.subsampling_rate = 6
        self.right_context = 10
        self.simple_down_2 = SpeechProjectionModule(odim, odim, 2)

    def forward(self, x: torch.Tensor, x_mask: torch.Tensor, offset: int = 0
               ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.
        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // 6.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // 6.
            torch.Tensor: positional encoding
        """
        x = x.unsqueeze(1)  # (b, c, t, f)
        x = self.conv(x)
        b, c, t, f = x.size()
        x = self.linear(x.transpose(1, 2).contiguous().view(b, t, c * f))
        cnn_seq_mask = x_mask[:, :, 10::6].squeeze(1)
        x, seq_mask = self.simple_down_2(x, cnn_seq_mask)
        seq_mask = seq_mask.unsqueeze(1)
        x, pos_emb = self.pos_enc(x, offset)
        return self.norm(x), pos_emb, seq_mask

class Conv2dSubsampling8(BaseSubsampling):
    """Convolutional 2D subsampling (to 1/8 length).

    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.

    """
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, norm: bool = False, **kwargs):
        """Construct an Conv2dSubsampling8 object."""
        super().__init__()
        self.conv = torch.nn.Sequential(
            torch.nn.Conv2d(1, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
        )
        self.linear = torch.nn.Linear(odim * ((idim - 7) // 8), odim)
        self.pos_enc = pos_enc_class
        self.norm = torch.nn.LayerNorm(odim) if norm else torch.nn.Identity()
        self.subsampling_rate = 8
        # 14 = (3 - 1) * 1 + (3 - 1) * 2 + (3 - 1) * 4
        self.right_context = 14

    def forward(self, x: torch.Tensor, x_mask: torch.Tensor, offset: int = 0
               ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.

        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // 8.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // 8.
            torch.Tensor: positional encoding
        """
        x = x.unsqueeze(1)  # (b, c, t, f)
        x = self.conv(x)
        b, c, t, f = x.size()
        x = self.linear(x.transpose(1, 2).contiguous().view(b, t, c * f))
        x, pos_emb = self.pos_enc(x, offset)
        return self.norm(x), pos_emb, x_mask[:, :, 14::8]

class Conv2dSubsampling16(BaseSubsampling):
    """Convolutional 2D subsampling (to 1/16 length).

    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.
    """
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, norm: bool = False, **kwargs):
        """Construct a Conv2dSubsampling16 object."""
        super().__init__()
        self.conv = torch.nn.Sequential(
            torch.nn.Conv2d(1, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
        )
        # After 4 convs with (k=3, s=2, no pad), the freq dim becomes (idim - 15) // 16
        self.linear = torch.nn.Linear(odim * ((idim - 15) // 16), odim)
        self.pos_enc = pos_enc_class
        self.norm = torch.nn.LayerNorm(odim) if norm else torch.nn.Identity()
        self.subsampling_rate = 16
        # 30 = (3 - 1) * (1 + 2 + 4 + 8)
        self.right_context = 30

    def forward(self, x: torch.Tensor, x_mask: torch.Tensor, offset: int = 0
               ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.

        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // 16.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // 16.
            torch.Tensor: positional encoding
        """
        x = x.unsqueeze(1)  # (b, c, t, f)
        x = self.conv(x)
        b, c, t, f = x.size()
        x = self.linear(x.transpose(1, 2).contiguous().view(b, t, c * f))
        x, pos_emb = self.pos_enc(x, offset)
        return self.norm(x), pos_emb, x_mask[:, :, 30::16]


class Conv2dSubsampling16(BaseSubsampling):
    """Convolutional 2D subsampling (to 1/16 length).

    Args:
        idim (int): Input dimension.
        odim (int): Output dimension.
        dropout_rate (float): Dropout rate.
    """

    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module):
        """Construct a Conv2dSubsampling16 object."""
        super().__init__()
        self.conv = torch.nn.Sequential(
            torch.nn.Conv2d(1, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
            torch.nn.Conv2d(odim, odim, 3, 2),
            torch.nn.ReLU(),
        )
        self.linear = torch.nn.Linear(
            odim * (((((idim - 1) // 2 - 1) // 2 - 1) // 2 - 1) // 2), odim
        )
        self.pos_enc = pos_enc_class
        self.subsampling_rate = 16
        # 30 = (3 - 1) * (1 + 2 + 4 + 8)
        self.right_context = 30

    def forward(
        self,
        x: torch.Tensor,
        x_mask: torch.Tensor,
        offset: int = 0
    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.

        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // 16.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // 16.
            torch.Tensor: positional encoding
        """
        x = x.unsqueeze(1)  # (b, c, t, f)
        x = self.conv(x)
        b, c, t, f = x.size()
        x = self.linear(x.transpose(1, 2).contiguous().view(b, t, c * f))
        x, pos_emb = self.pos_enc(x, offset)
        return x, pos_emb, x_mask[:, :, 30::16]





class HydraSubsampling(BaseSubsampling):
    def __init__(self, idim: int, odim: int, dropout_rate: float,
                 pos_enc_class: torch.nn.Module, subsampling_rates: str,
                 norm: bool = True, **kwargs):
        super().__init__()
        # NOTE(xcsong): example of subsampling_rates --> "4_6_8"
        print("subsample: {}".format(subsampling_rates))
        self.subsampling_rates = subsampling_rates.split('_')
        self.branches = torch.nn.ModuleDict()
        self.norms = torch.nn.ModuleDict()
        args = (idim, odim, dropout_rate, pos_enc_class)
        for rate in self.subsampling_rates:
            if norm:
                self.norms.update({rate: torch.nn.LayerNorm(odim)})
            else:
                self.norms.update({rate: torch.nn.Identity()})
            if rate == '1':
                self.branches.update({rate: LinearNoSubsampling(*args)})
            elif rate == '4':
                self.branches.update({rate: Conv2dSubsampling4(*args)})
            elif rate == '6':
                self.branches.update({rate: Conv2dSubsampling6(*args)})
            elif rate == '8':
                self.branches.update({rate: Conv2dSubsampling8(*args)})
            else:
                raise NotImplementedError("subsample-{} not impl.".format(rate))

    def forward(
        self,
        x: torch.Tensor,
        x_mask: torch.Tensor,
        offset: int = 0,
        subsampling_rate: str = "random",
    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Subsample x.

        Args:
            x (torch.Tensor): Input tensor (#batch, time, idim).
            x_mask (torch.Tensor): Input mask (#batch, 1, time).
            subsampling_rate (str): "random" / "4" / "6" / "8" ...,

        Returns:
            torch.Tensor: Subsampled tensor (#batch, time', odim),
                where time' = time // subsampling_rate.
            torch.Tensor: Subsampled mask (#batch, 1, time'),
                where time' = time // subsampling_rate.
            torch.Tensor: positional encoding
        """
        if subsampling_rate == "random":
            idx = torch.randint(0, len(self.subsampling_rates), (1,))
            subsampling_rate = self.subsampling_rates[idx[0]]
        print('SubSample for current batch: {}'.format(subsampling_rate))
        x, pos_emb, x_mask = self.branches[subsampling_rate](x, x_mask, offset)
        return self.norms[subsampling_rate](x), pos_emb, x_mask
